package edu.vt.ece.rt.quorum;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.jgroups.Address;
import org.jgroups.Global;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
import org.jgroups.conf.ConfiguratorFactory;
import org.jgroups.conf.ProtocolStackConfigurator;
import org.jgroups.protocols.FD;
import org.jgroups.protocols.FD_ALL;
import org.jgroups.protocols.pbcast.NAKACK;
import org.jgroups.stack.Protocol;
import org.jgroups.stack.ProtocolStack;

import edu.vt.ece.rt.quorum.messsages.MessageHandler;
import edu.vt.ece.rt.quorum.tree.Node;
import edu.vt.ece.rt.quorum.tree.Tree;
import edu.vt.ece.rt.quorum.util.Names;
import edu.vt.ece.rt.util.Logger.Logger;

/*
 * Basic node class allowing the node to communicate with other node
 * 
 * @author Sudhanshu 
 */

public class Cluster extends ReceiverAdapter{
	
	/*
	 * base Channel allows to create a cluster of all the nodes, helpful in detecting
	 * any node failure across the cluster.
	 */
	JChannel channel;
	
	/*
	 * Local node address 
	 */
	Address myAddress;
	
	/*
	 * All the members of cluster
	 */
	View currentView;
	
	/*
	 * Name of multicast group
	 */
	String group;
	
	/*
	 * Tree structure to create quorums from
	 */
	Tree tree;
	
	/*
	 * Node List, to access the quorums as List
	 */
	List<Node> nodeList;
	/*
	 * To verify whether it is mapped to Singleton Instance
	 */
	boolean sglt = false;
	
	public Cluster(){
		group = "Base";
	}
	
	public Cluster(String s){
		group = s;
	}
	
	/*
	 * To initiate a node and make it part of cluster "Quorum"
	 */
	public void init() throws Exception{
		ProtocolStackConfigurator configs = ConfiguratorFactory.getStackConfigurator("udp.xml");
		configs.getProtocolStack().get(0).getProperties().put(Global.SINGLETON_NAME, "instance");
				
		channel = new JChannel(configs);
		
		for (String keys : (ProtocolStack.getSingletonTransports()).keySet()){
			System.out.println("Singleton "+keys);
		}
		
		ProtocolStack stk = channel.getProtocolStack();		
			
		FD_ALL p1 = (FD_ALL) stk.findProtocol("FD_ALL");
		((FD) stk.findProtocol("FD")).stopFailureDetection();

		Protocol p = stk.findProtocol("UDP");
		if(p!=null){
			p.setLevel(System.getProperty(Names.jLogger));
		}
		
		channel.setReceiver(this);
        channel.connect(group);			//Add to multicast group Quorum
        channel.setDiscardOwnMessages(true);		//By Default no self receive of messages
        channel.getState(null, 10000);
        myAddress = channel.getAddress();   
        Logger.debug("My address "+myAddress);
        Thread.sleep(1000);
	}

	/*
	 * To wait on other nodes in network to get initiated
	 */
	public void waitForCluster(){
		waitForCluster(Integer.getInteger(Names.nodes));
	}
	public void waitForCluster(Integer nodes) {
			Logger.debug("Waiting for others to join "+group);
			while(currentView.getMembers().size() != nodes){
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			
			Logger.debug("<<<Quorum "+group+" ready>>>");
			if(group.equals(Names.base)){
				setNodeId();
				ClustersManager.startSyncr = new AtomicInteger(Integer.getInteger(Names.nodes));
			}
			
			try {
				Thread.sleep(3000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
	}

	private void setNodeId() {
		//Set the Node ID
		int i=0;
		for(Address adr: currentView.getMembers()){
			if(adr.equals(myAddress))
				break;
			i++;
		}
		ClustersManager.baseClusterId = i;
		Logger.debug("My ID "+i);
	}

	/*
	 *  To Close channel gracefully
	 */
	public void close(){
		channel.close();
	}

	@Override
	public void viewAccepted(View paramView) {
		currentView = paramView;		
		/*
		 * Check for reductions in members, if yes, trigger re-run of quorum protocol
		 * TODO: Other way, use the mute message to stop the node to talk other
		 */
	}
	
	public void createTree(int degree) {
		nodeList = new ArrayList<Node>();	
		for(Address adr: currentView.getMembers()){
			nodeList.add(new Node(adr));
		}
		ArrayList<Node> tmp = new ArrayList<Node>();
		for(Node n: nodeList)
			tmp.add(n);
		
		tree = Tree.createTree(tmp, degree);
	} 
	
	/*
	 * Depending on message it is send to group or unicast to particular member
	 */
	public void send(Message msg) throws Exception{
		channel.send(msg);
	}
	
	/*
	 * (non-Javadoc)
	 * @see org.jgroups.ReceiverAdapter#receive(org.jgroups.Message)
	 */
    public void receive(Message msg) {
    	MessageHandler.process(msg);
    }

    public int getReplyCount(){
    	//NOTE: Might me incorrect if implementation differ
    	return currentView.getMembers().size();
    }
    
    public Tree getTree(){
    	if(tree == null){
    		createTree(Integer.getInteger(Names.degree));
    	}
    	return tree;
    }
    
    public Address getMyAddress(){
    	return myAddress;
    }
    
    public void setAutoReceive(boolean b)	{
    	channel.setDiscardOwnMessages(!b);
    }
    
    public boolean getAutoReceive(){
    	return channel.getDiscardOwnMessages();
    }
    
    /*
     * Test the Cluster class, TODO: crude way, create JUnit tests
     */
    public static void main(String[] args) throws Exception {
		Cluster c = new Cluster();
		
		c.init();
		
		System.out.println("Sending Message");
		c.setAutoReceive(true);
		c.send(new Message(null,"Hello there"));
		
		c.close();
	}
}
